package com.shenma2009.utils;

import com.shenma2009.vo.HbaseVo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestBody;


import java.io.IOException;
import java.util.ArrayList;

/**
 * @author 卢锦涛
 */
@Component
public class HbaseUtils {
    private Connection connection;
    HBaseAdmin admin;
    //配置文件
    //hbase:
    //  zookeeper:
    //    quorum: 10.160.23.242
    //    property:
    //      clientPort: 2181
    //zookeeper:
    //  znode:
    //    parent: /hbase
    public void init() throws IOException {
        Configuration configuration = HBaseConfiguration.create();

        //设置zookeeper的地址，可以有多个，以逗号分隔 如果连不上自己虚拟机的Ip在虚拟机端口后面加上   :2181
        //修改本机域名 host文件
        configuration.set("hbase.zookeeper.quorum","10.160.23.242");
        // 创建hbase的连接，这是一个分布式连接
        connection = ConnectionFactory.createConnection(configuration);
        //这个admin是管理table时使用的，比如说创建表
        admin = (HBaseAdmin) connection.getAdmin();
    }

    /**
     * @description 创建命名空间
     * @author 卢锦涛
     * @date 2023/5/30 10:07
     */

    public boolean createNameSpace(String namespace) throws IOException {
        this.init();
        boolean flag=true;
        try {
            admin.getNamespaceDescriptor(namespace);
            flag=false;
        } catch (IOException e) {
            // 创建命令空间描述建造者 => 设计师
            NamespaceDescriptor.Builder builder = NamespaceDescriptor.create(namespace);
            // 给命令空间添加需求
            builder.addConfiguration("author","andy");
            builder.addConfiguration("date","2023-05-30");

            // 使用 builder 构造出对应的添加完参数的对象 完成创建
            // 创建命名空间出现的问题 都属于本方法自身的问题 不应该抛出
            admin.createNamespace(builder.build());
        }
        return flag;
    }

    /**
     * @description 创建表，创建表只需要指定列族，不需要指定列
     * 其实用命令真的会更快，create 'user','info1','info2'
     * @author 卢锦涛
     * @date 2023/5/30 18:42
     */
    public boolean createTable(@RequestBody HbaseVo hbaseVo) throws IOException {
        this.init();
        TableName tableName = TableName.valueOf(hbaseVo.getNamespace(), hbaseVo.getTableName());
        //检查表是否存在
        if (admin.tableExists(tableName)){
            return false;
        }
        //构造一个表的描述
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

        // 创建表的描述
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(hbaseVo.getColumnFamily()));
        // 添加版本参数
        columnFamilyDescriptorBuilder.setMaxVersions(5);
        // 创建添加完参数的列族描述
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
        // 创建表
        admin.createTable(tableDescriptorBuilder.build());
        return true;
    }


    /**
     * @description 在表中插入数据 添加
     * @author 卢锦涛
     * @date 2023/5/30 10:08
     */
    public boolean addDataToTable(HbaseVo hbaseVo) throws IOException {
        this.init();
        TableName tableName = TableName.valueOf(hbaseVo.getNamespace(), hbaseVo.getTableName());
        // 获取 table
        Table table = connection.getTable(tableName);
        // 创建 put 对象
        Put put = new Put(Bytes.toBytes(hbaseVo.getRowKey()));
        // 给 put 对象添加数据
        put.addColumn(Bytes.toBytes(hbaseVo.getColumnFamily()),Bytes.toBytes(hbaseVo.getColumnName()),Bytes.toBytes(hbaseVo.getValue()));
        // 将对象写入对应的方法
        table.put(put);
        // 关闭 table
        table.close();
        return admin.tableExists(tableName);
    }

    /**
     * @description 获取列表数据
     * @author 卢锦涛
     * @date 2023/5/30 19:42
     * shell cmd: scan 'shenmazong:student'
     */
    public ArrayList<String> listTableRows(HbaseVo hbaseVo) throws IOException {
        this.init();
        // 1. 获取 table
        Table table = connection.getTable(TableName.valueOf(hbaseVo.getNamespace(), hbaseVo.getTableName()));
        // 2. 创建 scan 对象
        Scan scan = new Scan();
        // 如果此时直接调用 会直接扫描整张表
        // 添加参数 来控制扫描的数据 默认包含
        scan.withStartRow(Bytes.toBytes(hbaseVo.getStartRow()));
        // 默认不包含
        scan.withStopRow(Bytes.toBytes(hbaseVo.getStopRow()));
        ArrayList<String> list = new ArrayList<>();
        try {// 读取多行数据 获得 scanner
            ResultScanner scanner = table.getScanner(scan);
            // result 来记录一行数据 cell 数组
            // ResultScanner 来记录多行数据 result 的数组
            for (org.apache.hadoop.hbase.client.Result result : scanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    String value=new
                            String(CellUtil.cloneRow(cell)) + "-" + new
                            String(CellUtil.cloneFamily(cell)) + "-" + new
                            String(CellUtil.cloneQualifier(cell)) + "-" + new
                            String(CellUtil.cloneValue(cell));
                    System.out.print (value);
                    list.add(value);
                }
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        // 3. 关闭 table
        table.close();
        return list;
    }


    /**
     * @description 获取单个数据
     * @author 卢锦涛
     * @date 2023/5/30 19:49
     */
    public ArrayList<String> getOne(HbaseVo hbaseVo) throws IOException {
        this.init();
        // 1. 获取 table
        Table table = connection.getTable(TableName.valueOf(hbaseVo.getNamespace(), hbaseVo.getTableName()));
        // 2. 创建 get 对象
        Get get = new Get(Bytes.toBytes(hbaseVo.getRowKey()));
        // 如果直接调用 get 方法读取数据 此时读一整行数据
        // 如果想读取某一列的数据 需要添加对应的参数
        get.addColumn(Bytes.toBytes(hbaseVo.getColumnFamily()),Bytes.toBytes(hbaseVo.getColumnName()));
        // 设置读取数据的版本
        get.readAllVersions();
        //get.readVersions(get.getMaxVersions());
        // 读取数据 得到 result 对象
        Result result = table.get(get);
        // 处理数据
        Cell[] cells = result.rawCells();
        // 测试方法: 直接把读取的数据打印到控制台
        // 如果l是实际开发 需要再额外写方法 对应处理数据
        ArrayList<String> list = new ArrayList<>();
        for (Cell cell : cells) {
            // cell 存储数据比较底层
            String value = new String(CellUtil.cloneValue(cell));
            list.add(value);
            System.out.println(value);
        }
        // 关闭 table
        table.close();
        return list;
    }

    /**
     * @description 删除记录
     * @author 卢锦涛
     * @date 2023/5/30 10:48
     */
    public void removeRow(HbaseVo hbaseVo) throws IOException {
        this.init();
        // 1.获取 table
        Table table = connection.getTable(TableName.valueOf(hbaseVo.getNamespace(), hbaseVo.getTableName()));
        // 2.创建 Delete 对象
        Delete delete = new Delete(Bytes.toBytes(hbaseVo.getRowKey()));
        // 3.添加删除信息
        // 3.1 删除单个版本
        delete.addColumn(Bytes.toBytes(hbaseVo.getColumnFamily()), Bytes.toBytes(hbaseVo.getColumnName()));
        // 3.2 删除所有版本，多了一个s
        //delete.addColumns(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
        // 3.3 删除列族
        // delete.addFamily(Bytes.toBytes(family));
        // 3.删除数据
        table.delete(delete);
        // 5.关闭资源
        table.close();
    }


    /**
     * @description 删除表
     * @author 卢锦涛
     * @date 2023/5/30 10:53
     */
    public boolean deleteTable(HbaseVo hbaseVo) throws IOException {
        this.init();
        // 调用相关的方法删除表格
        try {
            // HBase 删除表格之前 一定要先标记表格为不可以
            TableName tableName = TableName.valueOf(hbaseVo.getNamespace(), hbaseVo.getTableName());
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * @description 删除命名空间
     * @author 卢锦涛
     * @date 2023/5/30 20:26
     */
    public void deleteNamespace(HbaseVo hbaseVo) throws IOException {
        this.init();
        admin.deleteNamespace(hbaseVo.getNamespace());
    }

}
